package demo.kafka.interceptor;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Header;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.Map;

public class DemoConsumerInterceptor implements ConsumerInterceptor<String, String> {

    private final Logger log = LoggerFactory.getLogger(DemoConsumerInterceptor.class);

    private static final String TRACE_ID_HEADER_KEY = "x-trace-id";

    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        // 在消息被返回给用户的poll()方法之前调用

        // 可以在这里过滤或修改记录
        // 例如，我们检查每条记录的追踪ID
        for (ConsumerRecord<String, String> record : records) {
            Header traceHeader = record.headers().lastHeader(TRACE_ID_HEADER_KEY);
            if (traceHeader != null) {
                String traceId = new String(traceHeader.value(), StandardCharsets.UTF_8);
                log.info("CONSUMER INTERCEPTOR: Record from topic {} has trace ID: {}",
                        record.topic(), traceId);

                // 计算消息生产到消费的延迟（近似）
                long latency = System.currentTimeMillis() - record.timestamp();
                log.info("CONSUMER INTERCEPTOR: Message latency is approx {} ms", latency);

            } else {
                log.warn("CONSUMER INTERCEPTOR: Record from topic {} is missing trace ID!",
                        record.topic());
                // 在实际项目中，可以在这里进行异常处理或路由到死信队列
            }
        }

        // 如果你返回一个空的ConsumerRecords，则用户不会收到任何消息（实现了过滤）
        // return new ConsumerRecords<>(Map.of());

        // 正常返回原始记录
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        // 在偏移量提交成功后调用
        log.info("CONSUMER INTERCEPTOR: Offsets committed: {}", offsets);
    }

    @Override
    public void close() {
        // 拦截器关闭时调用
        log.info("CONSUMER INTERCEPTOR: Closing interceptor.");
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // 初始化方法
        log.info("CONSUMER INTERCEPTOR: Configuring interceptor with props: {}", configs);
    }
}
